package com.examples.mysql2mysql.batch;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class JdbcSourceFunction extends RichSourceFunction<Tuple2<String, Integer>> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private volatile boolean isRunning = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        // 创建数据库连接
        connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/f_saas_cfg?useUnicode=true&characterEncoding=UTF-8&useAffectedRows=true&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8", "root", "123456");

        // 准备查询语句
        preparedStatement = connection.prepareStatement("select `name` ,`del` from test_table a");

    }

    @Override
    public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
        ResultSet resultSet = preparedStatement.executeQuery();

        while (resultSet.next() && isRunning) {
            String name = resultSet.getString("name");
            int del = resultSet.getInt("del");
            ctx.collect(new Tuple2<>(name, del));
        }

        resultSet.close();
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void close() throws Exception {
        super.close();

        // 关闭连接和预处理语句
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
